package drds.binlog.parse.mysql.dbsync;


import drds.binlog.binlog_event.LogEvent;
import drds.binlog.binlog_event.event.*;
import drds.binlog.binlog_event.event.TableMapLogEvent.ColumnInfo;
import drds.binlog.binlog_event.event.rows_log_event.DeleteRowsLogEvent;
import drds.binlog.binlog_event.event.rows_log_event.RowsLogEvent;
import drds.binlog.binlog_event.event.rows_log_event.UpdateRowsLogEvent;
import drds.binlog.binlog_event.event.rows_log_event.WriteRowsLogEvent;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.data_object.*;
import drds.binlog.event_filter.aviater.RegexFilter;
import drds.binlog.parse.BinlogParser;
import drds.binlog.parse.FieldMetaData;
import drds.binlog.parse.TableMetaData;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.exception.TableIdNotFoundException;
import drds.binlog.parse.mysql.ddl.Ddl;
import drds.binlog.parse.mysql.ddl.FastsqlDdlParser;
import drds.binlog.parse.mysql.ddl.SimpleDdlParser;
import drds.common.$;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.sql.Types;
import java.util.BitSet;
import java.util.List;

/**
 * 基于{@linkplain LogEvent}转化为Entry对象的处理
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class BinlogLogEventConverter extends AbstractLifeCycle implements BinlogParser<LogEvent>
{

    public static final String XA_XID = "XA_XID";
    public static final String XA_TYPE = "XA_TYPE";
    public static final String XA_START = "XA START";
    public static final String XA_END = "XA END";
    public static final String XA_COMMIT = "XA COMMIT";
    public static final String XA_ROLLBACK = "XA ROLLBACK";
    public static final String ISO_8859_1 = "ISO-8859-1";
    public static final String UTF_8 = "UTF-8";
    public static final int TINYINT_MAX_VALUE = 256;
    public static final int SMALLINT_MAX_VALUE = 65536;
    public static final int MEDIUMINT_MAX_VALUE = 16777216;
    public static final long INTEGER_MAX_VALUE = 4294967296L;
    public static final BigInteger BIGINT_MAX_VALUE = new BigInteger("18446744073709551616");
    public static final int version = 1;
    public static final String BEGIN = "BEGIN";
    public static final String COMMIT = "COMMIT";

    @Setter
    @Getter
    private volatile RegexFilter schemaNameAndTableNameRegexFilter;                                                          // 运行时引用可能会有变化，比如规则发生变化时
    @Setter
    @Getter
    private volatile RegexFilter blackSchemaNameAndTableNameRegexFilter;


    @Setter
    @Getter
    private TableMetaDataCache tableMetaDataCache;
    @Setter
    @Getter
    private Charset connectionCharset = Charset.defaultCharset();
    @Setter
    @Getter
    private boolean filterQueryDcl = false;
    @Setter
    @Getter
    private boolean filterQueryDml = false;
    @Setter
    @Getter
    private boolean filterQueryDdl = false;
    // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
    @Setter
    @Getter
    private boolean filterTableError = false;
    // 新增rows过滤，用于仅订阅除rows以外的数据
    @Setter
    @Getter
    private boolean filterRows = false;
    @Setter
    @Getter
    private boolean useFastsqlDdlFilter = true;

    public BinlogLogEventConverter()
    {

    }

    public static TransactionBegin createTransactionBegin(long threadId)
    {
        TransactionBegin transactionBegin = new TransactionBegin();
        transactionBegin.setThreadId(threadId);
        return transactionBegin;
    }

    public static TransactionEnd createTransactionEnd(long transactionId)
    {
        TransactionEnd transactionEnd = new TransactionEnd();
        transactionEnd.setTransactionId(String.valueOf(transactionId));
        return transactionEnd;
    }

    public static KeyValuePair createPair(String key, String value)
    {
        KeyValuePair keyValuePair = new KeyValuePair();
        keyValuePair.setKey(key);
        keyValuePair.setValue(value);
        return keyValuePair;
    }

    public static Entry createEntry(Header header, EntryType entryType, DataObject dataObject)
    {
        Entry entry = new Entry();
        entry.setHeader(header);
        entry.setEntryType(entryType);
        entry.setDataObject(dataObject);
        return entry;
    }

    @Override
    public Entry parse(LogEvent logEvent, boolean isSeek) throws ParseException
    {
        if (logEvent == null || logEvent instanceof UnknownLogEvent)
        {
            return null;
        }

        int eventType = logEvent.getLogHeader().getLogType();
        switch (eventType)
        {
            case LogEvent.query_event:
                return parseQueryEvent((QueryLogEvent) logEvent, isSeek);
            case LogEvent.xid_event:
                return parseXidEvent((XidLogEvent) logEvent);
            case LogEvent.table_map_event:
                break;
            case LogEvent.write_rows_event_v1:
            case LogEvent.write_rows_event:
                return parseRowsEvent((WriteRowsLogEvent) logEvent);
            case LogEvent.update_rows_event_v1:
            case LogEvent.update_rows_event:
                return parseRowsEvent((UpdateRowsLogEvent) logEvent);
            case LogEvent.delete_rows_event_v1:
            case LogEvent.delete_rows_event:
                return parseRowsEvent((DeleteRowsLogEvent) logEvent);
            case LogEvent.rows_query_log_event:
                return parseRowsQueryEvent((RowsQueryLogEvent) logEvent);
            case LogEvent.annotate_rows_event:
                return parseAnnotateRowsEvent((AnnotateRowsEvent) logEvent);
            case LogEvent.user_var_event:
                return parseUserVarLogEvent((UserVarLogEvent) logEvent);
            case LogEvent.intvar_event:
                return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
            case LogEvent.rand_event:
                return parseRandLogEvent((RandLogEvent) logEvent);
            case LogEvent.gtid_log_event:
                return parseGTIDLogEvent((GtidLogEvent) logEvent);
            case LogEvent.heartbeat_log_event:
                return parseHeartbeatLogEvent();
            default:
                break;
        }

        return null;
    }

    public void reset()
    {
        // do nothing
        if (tableMetaDataCache != null)
        {
            tableMetaDataCache.clearTableMeta();
        }
    }

    private Entry parseHeartbeatLogEvent()
    {
        Header header = new Header();
        header.setEventType(EventType.master_heartbeat);
        Entry entry = new Entry();
        entry.setHeader(header);
        entry.setEntryType(EntryType.heartbeat);
        return entry;
    }

    private Entry parseGTIDLogEvent(GtidLogEvent gtidLogEvent)
    {
        LogHeader logHeader = gtidLogEvent.getLogHeader();
        KeyValuePair keyValuePair = new KeyValuePair();
        keyValuePair.setKey("gtid");
        keyValuePair.setValue(gtidLogEvent.getGtidString());

        if (gtidLogEvent.getLastCommitted() != null)
        {
            keyValuePair.setKey("lastCommitted");
            keyValuePair.setValue(String.valueOf(gtidLogEvent.getLastCommitted()));
            keyValuePair.setKey("sequenceNumber");
            keyValuePair.setValue(String.valueOf(gtidLogEvent.getSequenceNumber()));
        }

        Header header = createHeader(logHeader, "", "", EventType.gtid);
        return createEntry(header, EntryType.gtidlog, keyValuePair);
    }

    private Entry parseQueryEvent(QueryLogEvent queryLogEvent, boolean isSeek)
    {
        String queryString = queryLogEvent.getQuery();
        if (StringUtils.startsWithIgnoreCase(queryString, XA_START))
        {
            // xa start use TransactionBegin
            TransactionBegin transactionBegin = new TransactionBegin();
            transactionBegin.setThreadId(queryLogEvent.getSessionId());
            transactionBegin.getPropertyList().add(createPair(XA_TYPE, XA_START));
            transactionBegin.getPropertyList().add(createPair(XA_XID, getXaXid(queryString, XA_START)));
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_begin, transactionBegin);//need add , transactionBegin.toByteString()
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_END))
        {
            // xa start use TransactionEnd
            TransactionEnd transactionEnd = new TransactionEnd();
            transactionEnd.setTransactionId(String.valueOf(0L));
            transactionEnd.getPropertyList().add(createPair(XA_TYPE, XA_END));
            transactionEnd.getPropertyList().add(createPair(XA_XID, getXaXid(queryString, XA_END)));

            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_end, transactionEnd);
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT))
        {
            // xa commit
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", EventType.xa_commit);
            RowChange rowChange = new RowChange();
            rowChange.setSql(queryString);
            rowChange.getPropertyList().add(createPair(XA_TYPE, XA_COMMIT));
            rowChange.getPropertyList().add(createPair(XA_XID, getXaXid(queryString, XA_COMMIT)));
            rowChange.setEventType(EventType.xa_commit);
            return createEntry(header, EntryType.rowdata, rowChange);
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK))
        {
            // xa rollback
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", EventType.xa_rollback);
            RowChange rowChange = new RowChange();
            rowChange.setSql(queryString);
            rowChange.getPropertyList().add(createPair(XA_TYPE, XA_ROLLBACK));
            rowChange.getPropertyList().add(createPair(XA_XID, getXaXid(queryString, XA_ROLLBACK)));
            rowChange.setEventType(EventType.xa_rollback);
            return createEntry(header, EntryType.rowdata, rowChange);
        } else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN))
        {
            TransactionBegin transactionBegin = createTransactionBegin(queryLogEvent.getSessionId());
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_begin, transactionBegin);
        } else if (StringUtils.endsWithIgnoreCase(queryString, COMMIT))
        {
            TransactionEnd transactionEnd = createTransactionEnd(0L); // MyISAM可能不会有xid事件
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_end, transactionEnd);
        } else
        {
            boolean notFilter = false;
            EventType eventType = EventType.query;
            String tableName = null;
            String schemaName = null;
            if (useFastsqlDdlFilter)
            {
                List<Ddl> ddlList = FastsqlDdlParser.parse(queryString, queryLogEvent.getSchemaName());
                for (Ddl ddl : ddlList)
                {
                    if (!processFilter(queryString, ddl))
                    {
                        // 只要有一个数据不进行过滤
                        notFilter = true;
                    }
                }
                if (ddlList.size() > 0)
                {
                    // 如果针对多行的DDL,只能取第一条
                    eventType = ddlList.get(0).getEventType();
                    schemaName = ddlList.get(0).getSchemaName();
                    tableName = ddlList.get(0).getTableName();
                }
            } else
            {
                Ddl ddl = SimpleDdlParser.parse(queryString, queryLogEvent.getSchemaName());
                if (!processFilter(queryString, ddl))
                {
                    notFilter = true;
                }

                eventType = ddl.getEventType();
                schemaName = ddl.getSchemaName();
                tableName = ddl.getTableName();
            }

            if (!notFilter)
            {
                // 如果是过滤的数据就不处理了
                return null;
            }

            if (!isSeek)
            {
                // 使用新的表结构元数据管理方式
                EntryPosition entryPosition = createEntryPosition(queryLogEvent.getLogHeader());
                tableMetaDataCache.apply(entryPosition, queryLogEvent.getSchemaName(), queryString, null);
            }

            Header header = createHeader(queryLogEvent.getLogHeader(), schemaName, tableName, eventType);
            RowChange rowChange = new RowChange();
            if (eventType != EventType.query && eventType != EventType.insert && eventType != EventType.update
                    && eventType != EventType.delete)
            {
                rowChange.setDdl(true);
            }
            rowChange.setSql(queryString);
            if ($.isNotNullAndNotEmpty(queryLogEvent.getSchemaName()))
            {// 可能为空
                rowChange.setDdlSchemaName(queryLogEvent.getSchemaName());
            }
            rowChange.setEventType(eventType);
            //need add , rowChangeBuider.build()
            return createEntry(header, EntryType.rowdata, rowChange);
        }
    }

    private String getXaXid(String queryString, String type)
    {
        return StringUtils.substringAfter(queryString, type);
    }

    private boolean processFilter(String queryString, Ddl ddl)
    {
        String schemaName = ddl.getSchemaName();
        String tableName = ddl.getTableName();
        // 更新下table metadata cache
        if (tableMetaDataCache != null && (ddl.getEventType() == EventType.alter || ddl.getEventType() == EventType.erase || ddl.getEventType() == EventType.rename))
        {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            for (Ddl renameResult = ddl; renameResult != null; renameResult = renameResult.getRenameTableResult())
            {
                String schemaName0 = renameResult.getSchemaName();
                String tableName0 = renameResult.getTableName();
                if ($.isNotNullAndNotEmpty(tableName0))
                {
                    // 如果解析到了正确的表信息，则根据全名进行清除
                    tableMetaDataCache.clearTableMeta(schemaName0, tableName0);
                } else
                {
                    // 如果无法解析正确的表信息，则根据schema进行清除
                    tableMetaDataCache.clearTableMetaWithSchemaName(schemaName0);
                }
            }
        }


        if (ddl.getEventType() == EventType.alter ||//
                ddl.getEventType() == EventType.erase || //
                ddl.getEventType() == EventType.create || //
                ddl.getEventType() == EventType.truncate || //
                ddl.getEventType() == EventType.rename || //
                ddl.getEventType() == EventType.create_index ||//
                ddl.getEventType() == EventType.drop_index)//
        { // 针对DDL类型

            if (filterQueryDdl)
            {
                return true;
            }

            //rename
            if ($.isNullOrEmpty(tableName) || (ddl.getEventType() == EventType.rename && $.isNullOrEmpty(ddl.getOriginalTableName())))
            {
                // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析
                throw new ParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: " + queryString + " , and Ddl: " + ddl.toString());
                // return null;
            } else
            {
                // check name schemaNameAndTableNameRegexFilter
                String schemaNameAndTableName = schemaName + "." + tableName;
                if (schemaNameAndTableNameRegexFilter != null && !schemaNameAndTableNameRegexFilter.match(schemaNameAndTableName))
                {
                    if (ddl.getEventType() == EventType.rename)
                    {
                        // rename校验只要源和目标满足一个就进行操作
                        if (schemaNameAndTableNameRegexFilter != null && !schemaNameAndTableNameRegexFilter.match(ddl.getOriginalSchemaName() + "." + ddl.getOriginalTableName()))
                        {
                            return true;
                        }
                    } else
                    {
                        // 其他情况返回null
                        return true;
                    }
                }

                if (blackSchemaNameAndTableNameRegexFilter != null && blackSchemaNameAndTableNameRegexFilter.match(schemaNameAndTableName))
                {
                    if (ddl.getEventType() == EventType.rename)
                    {
                        // rename校验只要源和目标满足一个就进行操作
                        if (blackSchemaNameAndTableNameRegexFilter != null
                                && blackSchemaNameAndTableNameRegexFilter.match(ddl.getOriginalSchemaName() + "." + ddl.getOriginalTableName()))
                        {
                            return true;
                        }
                    } else
                    {
                        // 其他情况返回null
                        return true;
                    }
                }
            }
        } else if (ddl.getEventType() == EventType.insert || ddl.getEventType() == EventType.update
                || ddl.getEventType() == EventType.delete)
        {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            if (filterQueryDml)
            {
                return true;
            }
        } else if (filterQueryDcl)
        {
            return true;
        }

        return false;
    }

    private Entry parseRowsQueryEvent(RowsQueryLogEvent rowsQueryLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }
        // mysql5.6支持，需要设置binlog-rows-query-log-events=1，可详细打印原始DML语句
        String queryString = null;
        try
        {
            queryString = new String(rowsQueryLogEvent.getRowsQuery().getBytes(ISO_8859_1), connectionCharset.name());
            String tableName = null;
            if (useFastsqlDdlFilter)
            {
                List<Ddl> ddlList = FastsqlDdlParser.parse(queryString, null);
                if (ddlList.size() > 0)
                {
                    tableName = ddlList.get(0).getTableName();
                }
            }

            return buildQueryEntry(queryString, rowsQueryLogEvent.getLogHeader(), tableName);
        } catch (UnsupportedEncodingException e)
        {
            throw new ParseException(e);
        }
    }

    private Entry parseAnnotateRowsEvent(AnnotateRowsEvent annotateRowsEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }
        // mariaDb支持，需要设置binlog_annotate_row_events=true，可详细打印原始DML语句
        String queryString = null;
        try
        {
            queryString = new String(annotateRowsEvent.getRowsQuery().getBytes(ISO_8859_1), connectionCharset.name());
            return buildQueryEntry(queryString, annotateRowsEvent.getLogHeader());
        } catch (UnsupportedEncodingException e)
        {
            throw new ParseException(e);
        }
    }

    private Entry parseUserVarLogEvent(UserVarLogEvent userVarLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(userVarLogEvent.getQuery(), userVarLogEvent.getLogHeader());
    }

    private Entry parseIntrvarLogEvent(IntvarLogEvent intvarLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(intvarLogEvent.getQuery(), intvarLogEvent.getLogHeader());
    }

    private Entry parseRandLogEvent(RandLogEvent randLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(randLogEvent.getQuery(), randLogEvent.getLogHeader());
    }

    private Entry parseXidEvent(XidLogEvent xidLogEvent)
    {
        TransactionEnd transactionEnd = createTransactionEnd(xidLogEvent.getXid());
        Header header = createHeader(xidLogEvent.getLogHeader(), "", "", null);
        return createEntry(header, EntryType.transaction_end, transactionEnd);
    }

    public TableMetaData parseRowsEventForTableMetaData(RowsLogEvent rowsLogEvent)
    {
        TableMapLogEvent tableMapLogEvent = rowsLogEvent.getTableMapLogEvent();
        if (tableMapLogEvent == null)
        {
            // tableId对应的记录不存在
            throw new TableIdNotFoundException("not found tableId:" + rowsLogEvent.getTableId());
        }
        String schemaNameAndTableName = tableMapLogEvent.getSchemaName() + "." + tableMapLogEvent.getTableName();
        //
        if (schemaNameAndTableNameRegexFilter != null && !schemaNameAndTableNameRegexFilter.match(schemaNameAndTableName))
        {
            return null;
        }
        if (blackSchemaNameAndTableNameRegexFilter != null && blackSchemaNameAndTableNameRegexFilter.match(schemaNameAndTableName))
        {
            return null;
        }
        //
        TableMetaData tableMetaData = null;
        EntryPosition entryPosition = createEntryPosition(rowsLogEvent.getLogHeader());
        if (tableMetaDataCache != null && tableMetaData == null)
        {
            tableMetaData = getTableMetaData(tableMapLogEvent.getSchemaName(), tableMapLogEvent.getTableName(), true, entryPosition);
            if (tableMetaData == null)
            {
                if (!filterTableError)
                {
                    throw new ParseException("not found [" + schemaNameAndTableName + "] in db , pls check!");
                }
            }
        }

        return tableMetaData;
    }

    public Entry parseRowsEvent(RowsLogEvent rowsLogEvent)
    {
        return parseRowsEvent(rowsLogEvent, null);
    }

    public Entry parseRowsEvent(RowsLogEvent rowsLogEvent, TableMetaData tableMetaData)
    {
        if (filterRows)
        {
            return null;
        }
        try
        {
            if (tableMetaData == null)
            { // 如果没有外部指定
                tableMetaData = parseRowsEventForTableMetaData(rowsLogEvent);
            }

            if (tableMetaData == null)
            {
                // 拿不到表结构,执行忽略
                return null;
            }

            EventType eventType = null;
            int logType = rowsLogEvent.getLogHeader().getLogType();
            if (LogEvent.write_rows_event_v1 == logType || LogEvent.write_rows_event == logType)
            {
                eventType = EventType.insert;
            } else if (LogEvent.update_rows_event_v1 == logType || LogEvent.update_rows_event == logType)
            {
                eventType = EventType.update;
            } else if (LogEvent.delete_rows_event_v1 == logType || LogEvent.delete_rows_event == logType)
            {
                eventType = EventType.delete;
            } else
            {
                throw new ParseException("unsupport event logType :" + rowsLogEvent.getLogHeader().getLogType());
            }

            RowChange rowChange = new RowChange();
            rowChange.setTableId(rowsLogEvent.getTableId());
            rowChange.setDdl(false);

            rowChange.setEventType(eventType);
            RowsLogBuffer rowsLogBuffer = rowsLogEvent.getRowsBuf(connectionCharset.name());
            BitSet columns = rowsLogEvent.getColumns();
            BitSet changeColumns = rowsLogEvent.getChangeColumns();

            boolean tableError = false;
            int rowsCount = 0;
            while (rowsLogBuffer.nextOneRow(columns))
            {
                // 处理row记录
                RowData rowData = new RowData();
                if (EventType.insert == eventType)
                {
                    // insert的记录放在before字段中
                    tableError |= parseOneRow(rowData, rowsLogEvent, rowsLogBuffer, columns, true, tableMetaData);
                } else if (EventType.delete == eventType)
                {
                    // delete的记录放在before字段中
                    tableError |= parseOneRow(rowData, rowsLogEvent, rowsLogBuffer, columns, false, tableMetaData);
                } else
                {
                    // update需要处理before/after
                    tableError |= parseOneRow(rowData, rowsLogEvent, rowsLogBuffer, columns, false, tableMetaData);
                    if (!rowsLogBuffer.nextOneRow(changeColumns))
                    {
                        rowChange.getRowDataList().add(rowData);
                        break;
                    }

                    tableError |= parseOneRow(rowData, rowsLogEvent, rowsLogBuffer, changeColumns, true, tableMetaData);
                }

                rowsCount++;
                rowChange.getRowDataList().add(rowData);
            }
            TableMapLogEvent tableMapLogEvent = rowsLogEvent.getTableMapLogEvent();
            Header header = createHeader(rowsLogEvent.getLogHeader(),
                    tableMapLogEvent.getSchemaName(),
                    tableMapLogEvent.getTableName(),
                    eventType,
                    rowsCount);

            RowChange $rowChange = rowChange;
            if (tableError)
            {
                //need add , ByteString.EMPTY
                Entry entry = createEntry(header, EntryType.rowdata, $rowChange);
                log.warn("table parser error : {}storeValue: {}", entry.toString(), $rowChange.toString());
                return null;
            } else
            {
                //need add , rowChange.toByteString()
                Entry entry = createEntry(header, EntryType.rowdata, $rowChange);
                return entry;
            }
        } catch (Exception e)
        {
            throw new ParseException("parse row data failed.", e);
        }
    }

    private EntryPosition createEntryPosition(LogHeader logHeader)
    {
        return new EntryPosition(//
                logHeader.getServerId(), //
                logHeader.getLogFileName(),//
                logHeader.getLogPosition(),//
                logHeader.getCreateDateTimeLong() * 1000L//
        ); // 记录到秒
    }

    private boolean parseOneRow(RowData rowData, RowsLogEvent rowsLogEvent, RowsLogBuffer rowsLogBuffer, BitSet cols,
                                boolean isAfter, TableMetaData tableMetaData) throws UnsupportedEncodingException
    {
        int columnCount = rowsLogEvent.getTableMapLogEvent().getColumnCount();
        ColumnInfo[] columnInfos = rowsLogEvent.getTableMapLogEvent().getColumnInfos();

        boolean tableError = false;
        // check table fileds count，只能处理加字段
        boolean existRDSNoPrimaryKey = false;
        if (tableMetaData != null && columnInfos.length > tableMetaData.getFieldMetaDataList().size())
        {
            if (tableMetaDataCache.isOnRDS())
            {
                // 特殊处理下RDS的场景
                List<FieldMetaData> primaryKeys = tableMetaData.getPrimaryKeyFieldMetaDataList();
                if (primaryKeys == null || primaryKeys.isEmpty())
                {
                    if (columnInfos.length == tableMetaData.getFieldMetaDataList().size() + 1
                            && columnInfos[columnInfos.length - 1].type == LogEvent.mysql_type_longlong)
                    {
                        existRDSNoPrimaryKey = true;
                    }
                }
            }

            EntryPosition entryPosition = createEntryPosition(rowsLogEvent.getLogHeader());
            if (!existRDSNoPrimaryKey)
            {
                // online ddl增加字段操作步骤：
                // 1. 新增一张临时表，将需要做ddl表的数据全量导入
                // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
                // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
                // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
                // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
                tableMetaData = getTableMetaData(rowsLogEvent.getTableMapLogEvent().getSchemaName(), rowsLogEvent.getTableMapLogEvent().getTableName(), false, entryPosition);// 强制重新获取一次
                if (tableMetaData == null)
                {
                    tableError = true;
                    if (!filterTableError)
                    {
                        throw new ParseException("not found [" + rowsLogEvent.getTableMapLogEvent().getSchemaName() + "."
                                + rowsLogEvent.getTableMapLogEvent().getTableName() + "] in db , pls check!");
                    }
                }

                // 在做一次判断
                if (tableMetaData != null && columnInfos.length > tableMetaData.getFieldMetaDataList().size())
                {
                    tableError = true;
                    if (!filterTableError)
                    {
                        throw new ParseException("column size is not match for table:" + tableMetaData.getFullName()
                                + "," + columnInfos.length + " vs " + tableMetaData.getFieldMetaDataList().size());
                    }
                }
            } else
            {
                log.warn("[" + rowsLogEvent.getTableMapLogEvent().getSchemaName() + "." + rowsLogEvent.getTableMapLogEvent().getTableName()
                        + "] is no primary key , skip alibaba_rds_row_id column");
            }
        }

        for (int i = 0; i < columnCount; i++)
        {
            ColumnInfo columnInfo = columnInfos[i];
            // mysql 5.6开始支持nolob/mininal类型,并不一定记录所有的列,需要进行判断
            if (!cols.get(i))
            {
                continue;
            }

            if (existRDSNoPrimaryKey && i == columnCount - 1 && columnInfo.type == LogEvent.mysql_type_longlong)
            {
                // 不解析最后一列
                rowsLogBuffer.nextValue(columnInfo.type, columnInfo.meta, false);
                continue;
            }

            Column column = new Column();

            FieldMetaData fieldMetaData = null;
            if (tableMetaData != null && !tableError)
            {
                // 处理file metadata
                fieldMetaData = tableMetaData.getFieldMetaDataList().get(i);
                column.setName(fieldMetaData.getColumnName());
                column.setKey(fieldMetaData.isPrimaryKey());
                // 增加mysql type类型,issue 73
                column.setMysqlType(fieldMetaData.getColumnType());
            }
            column.setIndex(i);
            column.setNull(false);

            // fixed issue
            // https://github.com/alibaba/canal/issues/66，特殊处理binary/varbinary，不能做编码处理
            boolean isBinary = false;
            boolean isSingleBit = false;
            if (fieldMetaData != null)
            {
                if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "VARBINARY"))
                {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "BINARY"))
                {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "TINYINT(1)"))
                {
                    isSingleBit = true;
                }
            }
            rowsLogBuffer.nextValue(columnInfo.type, columnInfo.meta, isBinary);
            if (existRDSNoPrimaryKey && i == columnCount - 1 && columnInfo.type == LogEvent.mysql_type_longlong)
            {
                // 不解析最后一列
                continue;
            }

            int javaType = rowsLogBuffer.getJavaType();
            if (isSingleBit && javaType == Types.TINYINT)
            {
                javaType = Types.BIT;
            }
            if (rowsLogBuffer.isNull())
            {
                column.setNull(true);
            } else
            {
                final Serializable value = rowsLogBuffer.getValue();
                // 处理各种类型
                switch (javaType)
                {
                    case Types.INTEGER:
                    case Types.TINYINT:
                    case Types.SMALLINT:
                    case Types.BIGINT:
                        // 处理unsigned类型
                        Number number = (Number) value;
                        if (fieldMetaData != null && fieldMetaData.isUnsigned() && number.longValue() < 0)
                        {
                            switch (rowsLogBuffer.getLength())
                            {
                                case 1: /* MYSQL_TYPE_TINY */
                                    column.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.SMALLINT; // 往上加一个量级
                                    break;

                                case 2: /* MYSQL_TYPE_SHORT */
                                    column.setValue(String.valueOf(Integer.valueOf(SMALLINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 3: /* MYSQL_TYPE_INT24 */
                                    column.setValue(String.valueOf(Integer.valueOf(MEDIUMINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 4: /* MYSQL_TYPE_LONG */
                                    column.setValue(String.valueOf(Long.valueOf(INTEGER_MAX_VALUE
                                            + number.longValue())));
                                    javaType = Types.BIGINT; // 往上加一个量级
                                    break;

                                case 8: /* MYSQL_TYPE_LONGLONG */
                                    column.setValue(BIGINT_MAX_VALUE.add(BigInteger.valueOf(number.longValue()))
                                            .toString());
                                    javaType = Types.DECIMAL; // 往上加一个量级，避免执行出错
                                    break;
                            }
                        } else
                        {
                            // 对象为number类型，直接valueof即可
                            column.setValue(String.valueOf(value));
                        }
                        break;
                    case Types.REAL: // float
                    case Types.DOUBLE: // double
                        // 对象为number类型，直接valueof即可
                        column.setValue(String.valueOf(value));
                        break;
                    case Types.BIT:// bit
                        // 对象为number类型
                        column.setValue(String.valueOf(value));
                        break;
                    case Types.DECIMAL:
                        column.setValue(((BigDecimal) value).toPlainString());
                        break;
                    case Types.TIMESTAMP:
                        // 修复时间边界值
                        // String v = value.toString();
                        // v = v.substring(0, v.length() - 2);
                        // columnBuilder.setValue(v);
                        // break;
                    case Types.TIME:
                    case Types.DATE:
                        // 需要处理year
                        column.setValue(value.toString());
                        break;
                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.LONGVARBINARY:
                        // fixed text encoding
                        // https://github.com/AlibabaTech/canal/issues/18
                        // mysql binlog中blob/text都处理为blob类型，需要反查table
                        // metadata，按编码解析text
                        if (fieldMetaData != null && isText(fieldMetaData.getColumnType()))
                        {
                            column.setValue(new String((byte[]) value, connectionCharset));
                            javaType = Types.CLOB;
                        } else
                        {
                            // byte数组，直接使用iso-8859-1保留对应编码，浪费内存
                            column.setValue(new String((byte[]) value, ISO_8859_1));
                            // columnBuilder.setValueBytes(ByteString.copyFrom((byte[])
                            // value));
                            javaType = Types.BLOB;
                        }
                        break;
                    case Types.CHAR:
                    case Types.VARCHAR:
                        column.setValue(value.toString());
                        break;
                    default:
                        column.setValue(value.toString());
                }

            }

            column.setSqlType(javaType);
            // 设置是否update的标记位
            column.setUpdated(isAfter
                    && isUpdate(rowData.getAfterColumnList(),
                    column.isNull() ? null : (String) column.getValue(),
                    i));
            if (isAfter)
            {
                rowData.getAfterColumnList().add(column);
            } else
            {
                rowData.getBeforeColumnList().add(column);
            }
        }

        return tableError;

    }

    private Entry buildQueryEntry(String queryString, LogHeader logHeader, String tableName)
    {
        Header header = createHeader(logHeader, "", tableName, EventType.query);
        RowChange rowChange = new RowChange();
        rowChange.setSql(queryString);
        rowChange.setEventType(EventType.query);
        return createEntry(header, EntryType.rowdata, rowChange);
    }

    private Entry buildQueryEntry(String queryString, LogHeader logHeader)
    {
        Header header = createHeader(logHeader, "", "", EventType.query);
        RowChange rowChange = new RowChange();
        rowChange.setSql(queryString);
        rowChange.setEventType(EventType.query);
        return createEntry(header, EntryType.rowdata, rowChange);
    }

    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType)
    {
        return createHeader(logHeader, schemaName, tableName, eventType, -1);
    }

    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType,
                                Integer rowsCount)
    {
        // header会做信息冗余,方便以后做检索或者过滤
        Header header = new Header();
        header.setVersion(version);
        header.setLogFileName(logHeader.getLogFileName());
        header.setLogFileOffset(logHeader.getLogPosition() - logHeader.getEventLength());
        header.setServerId(logHeader.getServerId());
        header.setServerCode(UTF_8);// 经过java输出后所有的编码为unicode
        header.setExecuteDateTimeLong(logHeader.getCreateDateTimeLong() * 1000L);

        if (eventType != null)
        {
            header.setEventType(eventType);
        }
        if (schemaName != null)
        {
            header.setSchemaName(schemaName);
        }
        if (tableName != null)
        {
            header.setTableName(tableName);
        }
        header.setEventLength(logHeader.getEventLength());
        // enable gtid setIndexAsOriginIndexAndOffset$originIndex
        if ($.isNotNullAndNotEmpty(logHeader.getGtidSetStr()))
        {
            header.setGtid(logHeader.getGtidSetStr());
        }
        // add current gtid
        if ($.isNotNullAndNotEmpty(logHeader.getCurrentGtid()))
        {
            KeyValuePair keyValuePair = createPair("curtGtid", logHeader.getCurrentGtid());
            header.getPropertyList().add(keyValuePair);
        }
        // add current gtid sequence no
        if ($.isNotNullAndNotEmpty(logHeader.getCurrentGtidSn()))
        {
            KeyValuePair keyValuePair = createPair("curtGtidSn", logHeader.getCurrentGtidSn());
            header.getPropertyList().add(keyValuePair);
        }

        // add current gtid last committed
        if ($.isNotNullAndNotEmpty(logHeader.getCurrentGtidLastCommit()))
        {
            KeyValuePair keyValuePair = createPair("curtGtidLct", logHeader.getCurrentGtidLastCommit());
            header.getPropertyList().add(keyValuePair);
        }

        // add rowsCount suppport
        if (rowsCount > 0)
        {
            KeyValuePair keyValuePair = createPair("rowsCount", String.valueOf(rowsCount));
            header.getPropertyList().add(keyValuePair);
        }
        return header;
    }

    private boolean isUpdate(List<Column> columnList, String newValue, int index)
    {
        if (columnList == null)
        {
            throw new ParseException("ERROR ## the bfColumns is null");
        }

        if (index < 0)
        {
            return false;
        }

        for (Column column : columnList)
        {
            if (column.getIndex() == index)
            {// 比较before / after的column setIndexAsOriginIndexAndOffset$originIndex
                if (column.isNull() && newValue == null)
                {
                    // 如果全是null
                    return false;
                } else if (newValue != null && (!column.isNull() && column.getValue().equals(newValue)))
                {
                    // fixed issue #135, old column is Null
                    // 如果不为null，并且相等
                    return false;
                }
            }
        }

        // 比如nolob/minial模式下,可能找不到before记录,认为是有变化
        return true;
    }

    private TableMetaData getTableMetaData(String schemaName, String tableName, boolean useCache, EntryPosition entryPosition)
    {
        return tableMetaDataCache.getTableMetaData(schemaName, tableName, useCache, entryPosition);
    }

    private boolean isText(String columnType)
    {
        return "text".equalsIgnoreCase(columnType) ||//
                "longtext".equalsIgnoreCase(columnType) ||//
                //
                "tinytext".equalsIgnoreCase(columnType) ||//
                "mediumtext".equalsIgnoreCase(columnType);//
    }


}
